// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/bitmex/feed_v1/processor.h"

#include <utility>

#include <glog/logging.h>
#include <rapidjson/document.h>
#include <rapidjson/error/en.h>

#include "coin/proto/coin_order_enums.pb.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bitmex/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::bitmex::feed_v1::impl {
namespace {
void ParseInstrument(
    const rapidjson::Value& instrument_json,
    base::feed::impl::FeedUpdateManager* upd_mgr,
    bool is_snapshot) {
  const std::string& native_symbol = instrument_json["symbol"].GetString();
  if (!upd_mgr->is_symbol_registered(native_symbol)) {
    return;
  }
  if (instrument_json.HasMember("timestamp")){
    auto ts = TimestampFromIso8601(instrument_json["timestamp"].GetString());
    if (instrument_json.HasMember("openInterest")) {
      auto* open_interest = upd_mgr->GetOpenInterestMutable();
      open_interest->timestamp = ts;
      open_interest->open_interest_qty = instrument_json["openInterest"].GetInt64();
      upd_mgr->PublishOpenInterestWithExchangeTimeCheck(native_symbol);
    }
    if (instrument_json.HasMember("markPrice")) {
      auto* mark_price = upd_mgr->GetMarkPriceMutable();
      mark_price->timestamp = ts;
      if (instrument_json["markPrice"].IsDouble()) {
        mark_price->mark_price = instrument_json["markPrice"].GetDouble();
      } else {
        mark_price->mark_price = 0;
      }
      upd_mgr->PublishMarkPriceWithExchangeTimeCheck(native_symbol);
    }
    if (instrument_json.HasMember("fundingRate") ||
        instrument_json.HasMember("indicativeFundingRate") ||
        instrument_json.HasMember("fundingTimestamp")) {
      auto* funding_rate = upd_mgr->GetFundingRateMutable();
      if (BitmexFeedUpdateParser::ParseFundingRate(
          instrument_json, funding_rate, ts, is_snapshot)) {
        upd_mgr->PublishFundingRateWithExchangeTimeCheck(native_symbol);
      }
    }
  }
}

inline coin::proto::OrderSide GetOrderSide(const std::string& side) {
  if (side == "Buy") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "Sell") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}
}  // namespace

void BitmexFeedProcessor::AddProduct(const IProduct& product, IFeedSubscriber* subscriber) {
  try {
    native_to_pi_.emplace(
        product.native_symbol(),
        std::unique_ptr<IProductHolder>(
            product_holder_manager_->GetProductHolder(
                product, product.resolve_timestamp())->Clone()));
  } catch (...) {
    // It could only happen to BTC/ETH-USDT.NEXT_QUARTER.
    LOG(ERROR) << "[Bitmex Feed Processor] pi missing: " << product;
    return;
  }
  base::feed::FeedParsingProcessor::AddProduct(product, subscriber);
}

void BitmexFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;

  document.Parse(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error("JSON parse error: "s + GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("table")) {
    const std::string& table = document["table"].GetString();
    // parse this msg?
    // e.g. "trade:{}"
    if (!CheckChannelFilter(table + ":{}")) {
      return;
    }

    const int64_t timestamp = packet.timestamp;

    // TODO(donggu): orderBook10 has a different msg format, e.g. no 'id' field
    if (table == "orderBookL2" || table == "orderBookL2_25") {
      const std::string& action = document["action"].GetString();
      const auto arr = document["data"].GetArray();

      if (action == "partial") {
        for (rapidjson::SizeType i = 0; i < arr.Size();) {
          std::string native_symbol = arr[i]["symbol"].GetString();
          auto* book = upd_mgr_.GetBookMutable(native_symbol);
          if (book != nullptr) {
            if (publish_bbo_book_only_) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              i = parser_.ParseL2BookPartialItem(native_symbol, book, timestamp, arr, i);
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            } else {
              i = parser_.ParseL2BookPartialItem(native_symbol, book, timestamp, arr, i);
              upd_mgr_.PublishBook();
            }
          } else {
            // skip this symbol
            do {
              i++;
            } while (i < arr.Size() && arr[i]["symbol"].GetString() == native_symbol);
          }
        }
      } else if (action == "insert") {
        for (rapidjson::SizeType i = 0; i < arr.Size();) {
          std::string native_symbol = arr[i]["symbol"].GetString();
          auto* book = upd_mgr_.GetBookMutable(native_symbol);
          if (book != nullptr) {
            if (publish_bbo_book_only_) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              i = parser_.ParseL2BookInsertItem(
                  native_symbol,
                  book,
                  timestamp,
                  arr,
                  i,
                  publish_bbo_book_only_);
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            } else {
              i = parser_.ParseL2BookInsertItem(
                  native_symbol,
                  book,
                  timestamp,
                  arr,
                  i,
                  publish_bbo_book_only_);
              upd_mgr_.PublishBook();
            }
          } else {
            // skip this symbol
            do {
              i++;
            } while (i < arr.Size() && arr[i]["symbol"].GetString() == native_symbol);
          }
        }
      } else if (action == "delete") {
        bool discard = true;
        for (rapidjson::SizeType i = 0; i < arr.Size();) {
          std::string native_symbol = arr[i]["symbol"].GetString();
          auto* book = upd_mgr_.GetBookMutable(native_symbol);
          if (book != nullptr) {
            const auto& relative_norm = upd_mgr_.GetRelativeNorm(native_symbol);
            if (publish_bbo_book_only_) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              i = parser_.ParseL2BookDeleteItem(
                  native_symbol,
                  relative_norm,
                  book,
                  timestamp,
                  arr,
                  i,
                  &discard,
                  publish_bbo_book_only_,
                  native_to_pi_);
              if (discard) {
                continue;
              }
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            } else {
              i = parser_.ParseL2BookDeleteItem(
                  native_symbol,
                  relative_norm,
                  book,
                  timestamp,
                  arr,
                  i,
                  &discard,
                  publish_bbo_book_only_,
                  native_to_pi_);
              if (discard) {
                continue;
              }
              upd_mgr_.PublishBook();
            }
          } else {
            // skip this symbol
            do {
              i++;
            } while (i < arr.Size() && arr[i]["symbol"].GetString() == native_symbol);
          }
        }
      } else if (action == "update") {
        bool discard = true;
        for (rapidjson::SizeType i = 0; i < arr.Size();) {
          std::string native_symbol = arr[i]["symbol"].GetString();
          auto* book = upd_mgr_.GetBookMutable(native_symbol);
          if (book != nullptr) {
            const auto& relative_norm = upd_mgr_.GetRelativeNorm(native_symbol);
            if (publish_bbo_book_only_) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              i = parser_.ParseL2BookUpdateItem(
                  native_symbol,
                  relative_norm,
                  book,
                  timestamp,
                  arr,
                  i,
                  &discard,
                  publish_bbo_book_only_,
                  native_to_pi_);
              if (discard) {
                continue;
              }
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            } else {
              i = parser_.ParseL2BookUpdateItem(
                  native_symbol,
                  relative_norm,
                  book,
                  timestamp,
                  arr,
                  i,
                  &discard,
                  publish_bbo_book_only_,
                  native_to_pi_);
              if (discard) {
                continue;
              }
              upd_mgr_.PublishBook();
            }
          } else {
            // skip this symbol
            do {
              i++;
            } while (i < arr.Size() && arr[i]["symbol"].GetString() == native_symbol);
          }
        }
      }
    } else if (table == "orderBook10") {
      const std::string& action = document["action"].GetString();
      const auto arr = document["data"].GetArray();

      if (action == "partial") {
        for (rapidjson::SizeType i = 0; i < arr.Size();) {
          std::string native_symbol = arr[i]["symbol"].GetString();
          auto* book = upd_mgr_.GetBookMutable(native_symbol);
          if (book != nullptr) {
            if (publish_bbo_book_only_) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              i = parser_.Parse10BookPartialItem(native_symbol, book, timestamp, arr, i);
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            } else {
              i = parser_.Parse10BookPartialItem(native_symbol, book, timestamp, arr, i);
              upd_mgr_.PublishBook();
            }
          } else {
            // skip this symbol
            do {
              i++;
            } while (i < arr.Size() && arr[i]["symbol"].GetString() == native_symbol);
          }
        }
      } else if (action == "insert") {
        throw std::logic_error("this api is unexpected");
      } else if (action == "delete") {
        throw std::logic_error("this api is unexpected");
      } else if (action == "update") {
        for (rapidjson::SizeType i = 0; i < arr.Size();) {
          std::string native_symbol = arr[i]["symbol"].GetString();
          auto* book = upd_mgr_.GetBookMutable(native_symbol);
          if (book != nullptr) {
            if (publish_bbo_book_only_) {
              std::optional<PriceQty> bid_pre = book->Bid0();
              std::optional<PriceQty> ask_pre = book->Ask0();
              i = parser_.Parse10BookUpdateItem(
                  native_symbol,
                  book,
                  timestamp,
                  arr,
                  i,
                  publish_bbo_book_only_);
              std::optional<PriceQty> bid_post = book->Bid0();
              std::optional<PriceQty> ask_post = book->Ask0();
              bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
              if (is_bbo_changed) {
                upd_mgr_.PublishBook();
              }
            } else {
              i = parser_.Parse10BookUpdateItem(
                  native_symbol,
                  book,
                  timestamp,
                  arr,
                  i,
                  publish_bbo_book_only_);
              upd_mgr_.PublishBook();
            }
          } else {
            // skip this symbol
            do {
              i++;
            } while (i < arr.Size() && arr[i]["symbol"].GetString() == native_symbol);
          }
        }
      }
    } else if (table == "trade") {  // trade
      // trades: (similar to okex)
      //   symbol, ts, p, q, side
      //   symbol, ts, p, q, side
      //   symbol, ts, p, q, side
      // ...
      const std::string action = document["action"].GetString();
      if (action == "insert" || action == "partial") {
        const auto arr = document["data"].GetArray();

        std::string last_symbol = "";
        Trade* agg_trade = upd_mgr_.GetTradeMutable();
        agg_trade->Clear();
        Trade trade{};

        for (rapidjson::SizeType i = 0; i < arr.Size(); i++) {
          if (arr[i]["symbol"].GetString()[0] == '.') {  // index like ".XBT"
            continue;
          }
          // Settlement trade with no "BUY" or "SELL"
          if (!BitmexFeedUpdateParser::ParseTradeItem(arr[i], &trade)) {
            continue;
          }

          std::string native_symbol = arr[i]["symbol"].GetString();
          if (native_symbol == last_symbol && TryAggregateTrades(agg_trade, trade)) {
            continue;
          }
          if (agg_trade->fill_qty > 0) {
            CHECK_GT(last_symbol.size(), 0);
            upd_mgr_.PublishTrade(last_symbol, true);
          }
          *agg_trade = trade;
          last_symbol = std::move(native_symbol);
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(last_symbol, false);
        }
      } else {
        // trade + partial
        throw std::logic_error("bitmex unknown action: " + action);
      }
    } else if (table == "instrument") {
      const std::string action = document["action"].GetString();
      if (action == "partial") {
        for (const auto& item: document["data"].GetArray()) {
          ParseInstrument(item, &upd_mgr_, true);
        }
      } else if (action == "update") {
        for (const auto& item: document["data"].GetArray()) {
          ParseInstrument(item, &upd_mgr_, false);
        }
      } else {
        throw std::logic_error("bitmex instrument unknown action: " + action);
      }
    } else if (table == "liquidation") {
      const std::string action = document["action"].GetString();
      if (action == "insert") {
        for (const auto& order_obj : document["data"].GetArray()) {
          const std::string& native_symbol = order_obj["symbol"].GetString();
          if (!upd_mgr_.is_symbol_registered(native_symbol)) {
            continue;
          }
          auto* liquidation_order = upd_mgr_.GetLiquidationOrderMutable();
          liquidation_order->Clear();
          liquidation_order->price = order_obj["price"].GetDouble();
          liquidation_order->qty = order_obj["leavesQty"].GetInt64();
          liquidation_order->order_side = GetOrderSide(order_obj["side"].GetString());
          liquidation_order->order_id = order_obj["orderID"].GetString();
          liquidation_order->timestamp = packet.timestamp;
          native_to_liquidation_order_[native_symbol] = *liquidation_order;
          upd_mgr_.PublishLiquidationOrder(native_symbol, false);
        }
      } else if(action == "update") {
        for (const auto& order_obj : document["data"].GetArray()) {
          const std::string& native_symbol = order_obj["symbol"].GetString();
          if (!upd_mgr_.is_symbol_registered(native_symbol)) {
            continue;
          }
          if (!order_obj.HasMember("leavesQty")) {
            continue;
          }
          auto* liquidation_order = upd_mgr_.GetLiquidationOrderMutable();
          liquidation_order->Clear();
          if (native_to_liquidation_order_.count(native_symbol) == 1) {
            liquidation_order->order_side = native_to_liquidation_order_[native_symbol].order_side;
            liquidation_order->order_id = native_to_liquidation_order_[native_symbol].order_id;
            liquidation_order->price = native_to_liquidation_order_[native_symbol].price;
          } else {
            liquidation_order->order_side = coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
            liquidation_order->order_id = order_obj["orderID"].GetString();
          }
          if (!order_obj.HasMember("price") && liquidation_order->price == 0) {
            continue;
          }
          if (order_obj.HasMember("price")) {
            liquidation_order->price = order_obj["price"].GetDouble();
          }
          liquidation_order->qty = order_obj["leavesQty"].GetInt64();
          liquidation_order->timestamp = packet.timestamp;
          native_to_liquidation_order_[native_symbol] = *liquidation_order;
          upd_mgr_.PublishLiquidationOrder(native_symbol, false);
        }
      }
    }
  } else if (document.HasMember("url")) {
    const std::string url = document["url"].GetString();
    std::string channel;
    if (url.find("trade/bucketed") != std::string::npos) {
      channel = "kline_rest";
    }

    if (!CheckChannelFilter(channel)) {
      return;
    }

  if (channel == "kline_rest") {
      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      const std::string& native_symbol = parameters.at("symbol");
      auto interval = BitmexFeedUpdateParser::ParseKlineInterval(parameters.at("binSize"));
      auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (kline != nullptr) {
        bool to_publish = false;
        // query is split up several requests
        int page = -1;
        if (parameters.count("page") > 0) {
          page = std::stod(parameters["page"]);
        }
        if ((page == 0 ) || (page == -1 && document["data"].Size() > 10) ) {
          // first long query
          kline->SnapshotClear();
        } else {
          kline->UpdateClear();
        }

        for (const auto& item : document["data"].GetArray()) {
          bool freezed = BitmexFeedUpdateParser::Parse1Kline(
              item, kline, packet.timestamp, interval);
          to_publish = to_publish || freezed;
        }

        if (to_publish) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    }
  }

  upd_mgr_.onEndPacket(packet);
}

}  // namespace coin2::exchange::bitmex::feed_v1::impl
